package data

import (
	"context"

	protoV2 "Cosmos.Communication/api/v2"
	"Cosmos.Communication/internal/biz"
	"github.com/go-kratos/kratos/v2/log"
	"google.golang.org/protobuf/proto"
	"google.golang.org/protobuf/types/known/anypb"
	"google.golang.org/protobuf/types/known/wrapperspb"
)

type dataSet struct {
	infrastructure *Data
	logHelper      *log.Helper
}

func NewDataSet(data *Data, logger log.Logger) (d biz.DataSet, cf func(), err error) {
	instance := &dataSet{
		infrastructure: data,
	}
	instance.logHelper = log.NewHelper(logger)

	cf = instance.Close
	err = nil
	d = instance
	return
}

func (s *dataSet) Query(ctx context.Context, request *biz.DataQueryRequest) (result *biz.Result[*biz.DataQueryResponse], err error) {
	client, err := s.infrastructure.GetDataSetGrpcClient()
	if err != nil {
		return
	}

	result = &biz.Result[*biz.DataQueryResponse]{
		Code: 200,
	}
	rpcReq := &protoV2.RpcDataQueryRequest{
		Type:  &wrapperspb.StringValue{Value: request.Type},
		Param: &wrapperspb.StringValue{Value: request.Parameters},
	}
	rpcResult, err := client.Query(ctx, rpcReq)
	if err != nil {
		return
	}
	if rpcResult == nil {
		result.Code = 500
		result.Message = "Query RpcResult is nil"
		return
	}

	result = &biz.Result[*biz.DataQueryResponse]{
		Code:    int(rpcResult.Code),
		Message: rpcResult.Message.GetValue(),
		Data:    &biz.DataQueryResponse{},
	}
	if rpcResult.Data == nil {
		return
	}
	var rpcResponse = &protoV2.RpcDataQueryResponse{}
	err = anypb.UnmarshalTo(rpcResult.Data, rpcResponse, proto.UnmarshalOptions{})
	if err != nil {
		return
	}
	result.Data.Data = rpcResponse.Data.GetValue()
	return
}

func (s *dataSet) Subscribe(ctx context.Context, request *biz.DataSubscribeRequest) (result *biz.Result[*biz.DataSubscribeResponse], err error) {
	client, err := s.infrastructure.GetDataSetGrpcClient()
	if err != nil {
		return
	}

	result = &biz.Result[*biz.DataSubscribeResponse]{
		Code: 200,
	}
	rpcReq := &protoV2.RpcDataSubscribeRequest{
		Type:  &wrapperspb.StringValue{Value: request.Type},
		Param: &wrapperspb.StringValue{Value: request.Parameters},
	}
	rpcResult, err := client.Subscrib(ctx, rpcReq)
	if err != nil {
		return
	}
	if rpcResult == nil {
		result.Code = 500
		result.Message = "Subscribe RpcResult is nil"
		return
	}

	result = &biz.Result[*biz.DataSubscribeResponse]{
		Code:    int(rpcResult.Code),
		Message: rpcResult.Message.GetValue(),
		Data:    &biz.DataSubscribeResponse{},
	}
	if rpcResult.Data == nil {
		return
	}
	var rpcResponse = &protoV2.RpcDataSubscribeResponse{}
	err = anypb.UnmarshalTo(rpcResult.Data, rpcResponse, proto.UnmarshalOptions{})
	if err != nil {
		return
	}
	result.Data.Id = rpcResponse.Id.GetValue()
	return
}

func (s *dataSet) Unsubscribe(ctx context.Context, request *biz.DataUnsubscribeRequest) (result *biz.Result[*biz.DataUnsubscribeResponse], err error) {
	client, err := s.infrastructure.GetDataSetGrpcClient()
	if err != nil {
		return
	}

	result = &biz.Result[*biz.DataUnsubscribeResponse]{
		Code: 200,
	}
	rpcReq := &protoV2.RpcDataUnsubscribeRequest{
		Id: &wrapperspb.StringValue{Value: request.Id},
	}
	rpcResult, err := client.UnSubscrib(ctx, rpcReq)
	if err != nil {
		return
	}

	if rpcResult == nil {
		result.Code = 500
		result.Message = "Unsubscribe RpcResult is nil"
		return
	}
	result = &biz.Result[*biz.DataUnsubscribeResponse]{
		Code:    int(rpcResult.Code),
		Message: rpcResult.Message.GetValue(),
		Data:    &biz.DataUnsubscribeResponse{},
	}
	return
}

func (s *dataSet) Push(ctx context.Context, data *biz.DataFrame) (err error) {
	s.infrastructure.PushDataFrame(data)
	return nil
}

func (s *dataSet) Close() {
	s.infrastructure.Close()
}

func (d *dataSet) Poll() *biz.DataFrame {
	return nil
}
